from xtquant import xtdata
from datetime import datetime, timedelta
import time
import pandas as pd
from typing import Dict, Optional
from 主程.基础配置 import settings
from 组件.notice import show_notification_thread, init_notification_system
from 组件.logger import Logger
from 主程.指标.数据预热 import data_processor
from 主程.指标.日内九转 import IndependentNineTurnCalculator
from 主程.指标.波动率 import ATRCalculator
from 主程.指标.RSI import IndependentRSICalculator
from 主程.指标.VMA import IndependentVMACalculator
from 主程.指标.波动区间 import UpDownCalculator
from 组件.rabbitmq import RabbitMQTickPublisher
from 主程.指标.MACD import MACDSignalCalculator,MACDConfig, PerformanceMonitor, CacheConfig
from collections import deque

class BacktestMode:
    """回测模式类，用于模拟历史数据的tick推送"""
    
    def __init__(self, xtdata, stock_codes: list, trade_date: str, fast_mode: bool = True):
        """
        初始化回测模式
        
        Args:
            xtdata: 数据接口
            stock_codes: 要回测的股票代码列表
            trade_date: 回测日期，格式：YYYYMMDD
            fast_mode: 是否启用快速回测模式（默认True）
        """
        self.xtdata = xtdata
        self.stock_codes = stock_codes
        self.trade_date = trade_date
        self.trade_datetime = datetime.strptime(trade_date, "%Y%m%d")
        self.fast_mode = fast_mode
        
        # 存储每只股票的历史数据
        self.historical_data = {}
        self.current_positions = {}  # 当前数据位置
        self.is_running = False
        self.logger = Logger()
        
        self.logger.log_msg(f"🔄 回测模式初始化完成:")
        self.logger.log_msg(f"  - 回测日期: {trade_date}")
        self.logger.log_msg(f"  - 股票数量: {len(stock_codes)}")
        self.logger.log_msg(f"  - 股票列表: {', '.join(stock_codes)}")
        self.logger.log_msg(f"  - 快速模式: {'启用' if fast_mode else '禁用'}")

    def prepare_historical_data(self):
        """准备历史数据"""
        self.logger.log_msg("📊 正在准备回测历史数据...")

        for stock_code in self.stock_codes:
            try:
                # 首先尝试获取tick数据
                self.logger.log_msg(f"🔍 {stock_code} 尝试获取tick数据...")

                try:
                    self.xtdata.download_history_data(
                        stock_code=stock_code,
                        period="tick",
                        start_time=self.trade_date,
                        end_time=self.trade_date
                    )

                    data = self.xtdata.get_local_data(
                        stock_list=[stock_code],
                        period='tick',
                        start_time=self.trade_date,
                        end_time=self.trade_date
                    )

                    if data and stock_code in data and not data[stock_code].empty:
                        self.logger.log_msg(f"✅ {stock_code} tick数据获取成功")
                        data_source = "tick"
                    else:
                        self.logger.log_msg(f"⚠️ {stock_code} tick数据为空，尝试分钟级数据...")
                        data_source = None

                except Exception as tick_error:
                    self.logger.log_msg(f"⚠️ {stock_code} tick数据获取失败: {tick_error}")
                    data_source = None


                if data and stock_code in data:
                    df = data[stock_code]
                    self.logger.log_msg(f"🔍 {stock_code} 获取到数据，列名: {list(df.columns)}")
                    self.logger.log_msg(f"🔍 {stock_code} 数据形状: {df.shape}")
                    self.logger.log_msg(f"🔍 {stock_code} 前3行数据:")
                    self.logger.log_msg(df.head(3))

                    if not df.empty:
                        # 转换为标准tick格式
                        tick_data = self._convert_tick_data(df, stock_code)
                        if tick_data:  # 确保转换成功
                            self.historical_data[stock_code] = tick_data
                            self.current_positions[stock_code] = 0
                            self.logger.log_msg(f"✅ {stock_code} 历史数据准备完成: {len(tick_data)}个tick")
                        else:
                            self.logger.log_msg(f"⚠️ {stock_code} 数据转换失败")
                    else:
                        self.logger.log_msg(f"⚠️ {stock_code} 当日无交易数据")
                else:
                    self.logger.log_msg(f"❌ {stock_code} 无法获取历史数据")

            except Exception as e:
                self.logger.log_msg(f"❌ {stock_code} 数据准备失败: {e}")
                continue

        self.logger.log_msg(f"📊 历史数据准备完成，共 {len([k for k, v in self.historical_data.items() if v])} 只股票")

    def _convert_tick_data(self, df: pd.DataFrame, stock_code: str) -> list:
        """将tick数据转换为标准格式，只保留9:30-15:00的交易时段数据"""
        ticks = []

        # 定义交易时段
        morning_start = datetime.strptime("09:30:00", "%H:%M:%S").time()
        morning_end = datetime.strptime("11:30:00", "%H:%M:%S").time()
        afternoon_start = datetime.strptime("13:00:00", "%H:%M:%S").time()
        afternoon_end = datetime.strptime("15:00:00", "%H:%M:%S").time()

        for _, row in df.iterrows():
            try:
                # 解析时间
                tick_time = row.get('time') or row.get('Time') or row.get('datetime')
                if tick_time is None:
                    continue

                # 转换为datetime对象
                if isinstance(tick_time, (int, float)):
                    # 毫秒时间戳
                    dt = datetime.fromtimestamp(tick_time / 1000.0)
                elif isinstance(tick_time, str):
                    # 字符串时间
                    try:
                        dt = pd.to_datetime(tick_time).to_pydatetime()
                    except:
                        continue
                else:
                    dt = tick_time

                # 只保留交易时段的数据（9:30-11:30, 13:00-15:00）
                time_only = dt.time()
                if not ((morning_start <= time_only <= morning_end) or
                       (afternoon_start <= time_only <= afternoon_end)):
                    continue

                # 创建tick数据结构
                tick = {
                    'time': tick_time,
                    'lastPrice': float(row.get('lastPrice') or row.get('close') or row.get('Close') or row.get('price') or row.get('Price') or 0),
                    'open': float(row.get('open') or row.get('Open') or row.get('lastPrice') or 0),
                    'high': float(row.get('high') or row.get('High') or row.get('lastPrice') or 0),
                    'low': float(row.get('low') or row.get('Low') or row.get('lastPrice') or 0),
                    'close': float(row.get('close') or row.get('Close') or row.get('lastPrice') or 0),
                    'volume': int(row.get('volume') or row.get('Volume') or 0),
                    'amount': float(row.get('amount') or row.get('Amount') or 0),
                    'code': stock_code
                }
                ticks.append(tick)
            except Exception as e:
                print(f"⚠️ 跳过无效tick数据: {e}")
                continue

        # 按时间排序
        if ticks:
            try:
                ticks.sort(key=lambda x: x['time'])
            except:
                pass

        print(f"✅ {stock_code} 过滤后保留 {len(ticks)} 个交易时段tick数据")
        return ticks

    def get_next_tick_batch(self, batch_size: int = 1) -> Dict[str, list]:
        """获取下一批tick数据（支持批量获取）"""
        tick_batch = {}

        for stock_code in self.stock_codes:
            if stock_code in self.historical_data:
                data = self.historical_data[stock_code]
                current_pos = self.current_positions[stock_code]

                if current_pos < len(data):
                    # 计算本次要获取的tick数量
                    remaining = len(data) - current_pos
                    to_get = min(batch_size, remaining)

                    # 返回批量tick数据
                    tick_batch[stock_code] = data[current_pos:current_pos + to_get]
                    # 移动到下一个位置
                    self.current_positions[stock_code] += to_get

        return tick_batch

    def has_more_data(self) -> bool:
        """检查是否还有更多数据"""
        for stock_code in self.stock_codes:
            if stock_code in self.historical_data:
                current_pos = self.current_positions[stock_code]
                data_length = len(self.historical_data[stock_code])
                if current_pos < data_length:
                    return True
        return False

    def get_progress(self) -> Dict[str, float]:
        """获取回测进度"""
        progress = {}
        for stock_code in self.stock_codes:
            if stock_code in self.historical_data:
                current_pos = self.current_positions[stock_code]
                data_length = len(self.historical_data[stock_code])
                if data_length > 0:
                    progress[stock_code] = (current_pos / data_length) * 100
                else:
                    progress[stock_code] = 0
        return progress


class MarketMonitor:
    ##初始化
    def __init__(self, backtest_mode: bool = False, backtest_stocks: list = None, backtest_date: str = None):
        self.full_indicator_mode = True  # 如果为True则计算所有指标，否则只计算MACD
        self.backtest_mode = backtest_mode
        self.xtdata = xtdata  # 添加xtdata引用

        if backtest_mode:
            self.stock_codes = backtest_stocks
            self.backtest_date = backtest_date
            self._current_trade_date = backtest_date
            print(f"🔄 启动监控器已切换到回测模式")
            print(f"  - 回测日期: {backtest_date}")
            print(f"  - 回测股票: {', '.join(backtest_stocks)}")

        else:
            # 实时模式：使用配置文件中的股票代码
            self.stock_codes = [s["all_code"] for s in settings.STOCK_CODE]
            self.backtest_date = None
            self._current_trade_date = None

        self.period = "tick"  # 主数据流仍使用tick
        # RSI计算器现在完全独立，不需要订阅
        if backtest_mode:
            # 回测模式：为指定的股票创建名称映射
            self.stock_type = {}
            for stock in settings.STOCK_CODE:
                if stock["all_code"] in self.stock_codes:
                    self.stock_type[stock["all_code"]] = stock["name"]
            # 为没有配置的股票添加默认名称
            for code in self.stock_codes:
                if code not in self.stock_type:
                    self.stock_type[code] = f"回测股票_{code}"
        else:
            # 实时模式：使用配置文件中的股票代码
            self.stock_type = {stock["all_code"]: stock["name"] for stock in settings.STOCK_CODE}

        self._running = False
        self.saver = RabbitMQTickPublisher()

        # 九转重复买入保护时间（分钟）
        self.nine_turn_buy_protection_minutes = 10

        # 日内九转计算器（完全独立，直接获取历史数据计算，无需预热）
        self.nine_turn = IndependentNineTurnCalculator(
            compare_offset=4,
            required_bars=9,
            verbose=True,
            restrict_session=True,
            xtdata=xtdata
        )
        print(f"✅ 九转指标初始化完成: compare_offset=4, required_bars=9, verbose=True, restrict_session=True")
        print(f"✅ 九转重复买入保护时间: {self.nine_turn_buy_protection_minutes} 分钟")
        # ATR 波动率计算器（用于金叉买入时的波动过滤）
        self.atr_calculator = ATRCalculator(xtdata, lookback=settings.ATR_LOOKBACK, method='ema', cache_ttl_seconds=10, verbose=False)
        # RSI 计算器（完全独立，直接获取历史数据计算，无需预热）
        self.rsi_calculator = IndependentRSICalculator(period=5, xtdata=xtdata)
        # 区间波动计算器
        self.updown_calculator = UpDownCalculator()
        # 日志记录器
        self.logger = Logger()


        self._minute_bucket: Dict[str, str] = {}
        self._minute_close: Dict[str, float] = {}

        # VMA成交量指标（使用独立的VMA计算器）
        self.vma_calculator = IndependentVMACalculator(period=10, factor=1.2, xtdata=xtdata)

        # 九转信号时间保护：记录每个股票最后一次九转信号的时间
        self.nine_turn_signal_time = {code: {
            "last_buy_time": None,
            "last_sell_time": None
        } for code in self.stock_codes}

        ##macd配置
        self.MACD_DOWNLOAD_DELAY = 0.3  # MACD数据下载延迟（秒）
        # MACD相关初始化
        config = MACDConfig.get_optimized_params()
        self.macd_calculator = MACDSignalCalculator(self.xtdata, use_warmup=config['use_warmup'])
        self.macd_fast = MACDConfig.FAST_EMA
        self.macd_slow = MACDConfig.SLOW_EMA
        self.macd_signal = MACDConfig.SIGNAL_EMA
        self.tick_counters = {code: 0 for code in self.stock_codes}  # 每个股票的tick计数器
        self.macd_cache = {}  # MACD结果缓存，避免重复计算
        self.last_macd_check = {code: 0 for code in self.stock_codes}  # 上次MACD检查时间
        self.macd_check_interval = MACDConfig.TICK_CHECK_INTERVAL  # 每N个tick检查一次MACD
        self.macd_cache_ttl = config['cache_ttl']  # MACD缓存有效期（秒）
        self.max_cache_size = config['max_cache_size']  # 最大缓存大小

        # 添加信号状态跟踪，避免重复提示和下单
        self.signal_status = {code: {"last_golden_cross": None, "last_death_cross": None} for code in self.stock_codes}
        # 维护价格历史用于EMA计算（每只股票最近100个tick）
        self.price_history = {code: deque(maxlen=100) for code in self.stock_codes}



        # 回测模式相关初始化
        if backtest_mode:
            self.backtest_manager = BacktestMode(xtdata, self.stock_codes, backtest_date, True)
            print("✅ 已启用回测模式")
        else:
            self.backtest_manager = None


    ##订阅tick回调
    def quote_callback(self, data):
        for stock_code, stock_data in data.items():
            stock_name = self.stock_type.get(stock_code)

            # 取最新一笔：time 最大
            data0 = self._get_latest_tick_data(stock_data)

            last_price = data0.get('lastPrice')
            timestamp = data0.get('time')
            volume = data0.get('volume', 0)

            # 数据验证
            if last_price is None or timestamp is None:
                self.log(f"⚠️ {stock_name} 数据无效: price={last_price}, timestamp={timestamp}")
                continue
            # 增加tick计数器
            self.tick_counters[stock_code] += 1
            # 记录价格到历史用于EMA计算
            self.price_history[stock_code].append(last_price)

            # 计算所有指标
            indicators = self._calculate_all_indicators(stock_code, stock_name, last_price, timestamp, stock_data,volume)

            # 打印指标结果
            self._print_indicators(stock_code, stock_name, last_price, indicators, timestamp)

            # 将指标结果传给信号判断函数
            self.check_buy_sell(stock_code, stock_name, last_price, indicators)
    ##计算所有技术指标
    def _calculate_all_indicators(self, stock_code: str, stock_name: str, last_price: float, timestamp, stock_data, volume):
        """计算所有技术指标"""
        indicators = {}

        # 在回测模式下，计算当前tick对应的历史数据结束时间
        current_data_time = None
        if self.backtest_mode and timestamp is not None:
            try:
                if isinstance(timestamp, (int, float)):
                    # 尝试不同的时间戳格式
                    if timestamp > 1e10:  # 毫秒时间戳
                        current_data_time = datetime.fromtimestamp(timestamp / 1000.0)
                    else:  # 秒时间戳
                        current_data_time = datetime.fromtimestamp(timestamp)
                elif isinstance(timestamp, str):
                    current_data_time = pd.to_datetime(timestamp).to_pydatetime()
                else:
                    current_data_time = timestamp.to_pydatetime() if hasattr(timestamp, 'to_pydatetime') else timestamp
            except Exception as e:
                self.log(f"⚠️ {stock_code} 时间解析失败: {e}")
                current_data_time = None

        if self.full_indicator_mode:
            # 1. VMA成交量指标
            try:
                if self.backtest_mode and current_data_time:
                    indicators['vma_value'] = self.vma_calculator.get_latest_vma(stock_code, end_time=current_data_time)
                    indicators['volume_analysis'] = self.vma_calculator.get_volume_analysis(stock_code, end_time=current_data_time)
                else:
                    indicators['vma_value'] = self.vma_calculator.get_latest_vma(stock_code)
                    indicators['volume_analysis'] = self.vma_calculator.get_volume_analysis(stock_code)
            except Exception as e:
                indicators['vma_value'] = None
                indicators['volume_analysis'] = None
                self.log(f"⚠️ {stock_code} VMA处理异常: {e}")

            # 2. RSI指标
            try:
                # 在回测模式下，传递当前时间给RSI计算器，避免未来数据泄露
                if self.backtest_mode and current_data_time:
                    indicators['rsi'] = self.rsi_calculator.get_latest_rsi(stock_code, end_time=current_data_time)
                else:
                    indicators['rsi'] = self.rsi_calculator.get_latest_rsi(stock_code)
            except Exception as e:
                indicators['rsi'] = None
                self.log(f"⚠️ {stock_code} RSI获取异常: {e}")

            # 3. 日内九转指标（获取上一分钟的完整信号，避免当前分钟数据不完整）
            try:
                if self.backtest_mode and current_data_time:
                    nine_turn_signal = self.nine_turn.get_previous_minute_signal(stock_code,                                                             end_time=current_data_time)
                else:
                    nine_turn_signal = self.nine_turn.get_previous_minute_signal(stock_code)
                indicators['nine_turn'] = nine_turn_signal
            except Exception as e:
                self.log(f"⚠️ {stock_code} 九转信号获取异常: {e}")
                indicators['nine_turn'] = ""

            # 4. 波动区间指标
            if stock_code in ("513130.SH", "513061.SH"):
                try:
                    indicators['updown_id'] = self.updown_calculator.updown_watch_tick(stock_code, stock_data)
                except Exception as e:
                    self.log(f"⚠️ {stock_code} 波动区间计算异常: {e}")

            # 7. ATR波动率指标
            try:
                # 在回测模式下，传递当前时间给ATR计算器，避免未来数据泄露
                if self.backtest_mode and current_data_time:
                    atr = self.atr_calculator.calculate_atr(stock_code, last_price, end_time=current_data_time)
                else:
                    atr = self.atr_calculator.calculate_atr(stock_code, last_price)
                if atr is not None and atr > 0 and last_price:
                    indicators['atr_ratio'] = atr / last_price
            except Exception as e:
                self.log(f"⚠️ {stock_code} ATR计算异常: {e}")
                indicators['atr_ratio'] = None

            # 8. VMA指标（成交量放大判断）
            try:
                # 使用VMA计算器检查成交量是否放大
                # 在回测模式下，传递当前时间给VMA计算器，避免未来数据泄露
                if self.backtest_mode and current_data_time:
                    surge_result = self.vma_calculator.check_volume_surge(stock_code,
                                                                          end_time=current_data_time)
                else:
                    surge_result = self.vma_calculator.check_volume_surge(stock_code)
                indicators['vma'] = surge_result.get('is_surge', False)

                # 记录VMA判断结果
                if surge_result.get('is_surge'):
                    self.log(f"📈 {stock_code} VMA成交量放大: {surge_result.get('message', '')}")
                else:
                    # 可选：记录非放大状态（调试用）
                    if hasattr(self, '_debug_vma') and self._debug_vma:
                        self.log(f"📊 {stock_code} VMA状态: {surge_result.get('message', '')}")
                    # 记录简要状态（即使调试关闭也显示）
                    elif surge_result.get('message'):
                        self.log(f"📊 {stock_code} VMA: {surge_result.get('message', '')}")

            except Exception as e:
                self.log(f"⚠️ {stock_code} VMA判断异常: {e}")
                indicators['vma'] = False



        #  MACD指标
        if self.tick_counters[stock_code] % self.macd_check_interval == 0:
            macd = self.check_macd_signals(stock_code, last_price)

        return indicators
    ##打印所有指标结果
    def _print_indicators(self, stock_code: str, stock_name: str, last_price: float, indicators: dict, timestamp=None):
        """打印所有指标结果"""
        try:
            # 显示时间：回测模式使用数据时间，实时模式使用当前时间
            if self.backtest_mode and timestamp is not None:
                try:
                    # 尝试解析回测数据中的时间戳
                    if isinstance(timestamp, (int, float)):
                        # 如果是时间戳，转换为时间
                        data_time = datetime.fromtimestamp(timestamp / 1000.0)  # 假设是毫秒时间戳
                    elif isinstance(timestamp, str):
                        # 如果是字符串，尝试解析
                        data_time = pd.to_datetime(timestamp)
                    else:
                        # 如果是pandas时间戳，直接转换
                        data_time = timestamp.to_pydatetime() if hasattr(timestamp, 'to_pydatetime') else timestamp

                    # 格式化显示时间
                    if hasattr(data_time, 'strftime'):
                        display_time = data_time.strftime('%H:%M:%S')
                    else:
                        display_time = str(data_time)
                except Exception:
                    # 如果时间解析失败，使用当前时间
                    display_time = datetime.now().strftime('%H:%M:%S')
            else:
                # 实时模式使用当前时间
                display_time = datetime.now().strftime('%H:%M:%S')

            self.logger.log_msg(f"\n📊 [{display_time}] {stock_name}({stock_code}) - 价格: {last_price}")

            # 显示RSI指标
            rsi = indicators.get('rsi')
            rsi_str = f"{rsi:.2f}" if rsi is not None else "N/A"
            self.logger.log_msg(f"   RSI: {rsi_str}")

            # 显示VMA指标
            vma_value = indicators.get('vma_value')
            volume_analysis = indicators.get('volume_analysis')

            if vma_value and volume_analysis:
                current_vol = volume_analysis.get('current_volume', 0)
                last_valid_vol = volume_analysis.get('last_valid_volume', 0)
                volume_ratio = volume_analysis.get('volume_ratio', 0)
                trading_status = volume_analysis.get('trading_status', '未知')

                self.logger.log_msg(f"   VMA: {vma_value:.0f}")
                self.logger.log_msg(f"   成交量: 当前={current_vol:.0f}, 最近有效={last_valid_vol:.0f}, 倍数: {volume_ratio:.2f}, 状态: {trading_status}")
            else:
                self.logger.log_msg(f"   VMA: N/A")

            # 显示九转信号状态
            nine_turn_signal = indicators.get('nine_turn', '')
            if nine_turn_signal:
                self.logger.log_msg(f"   九转: {nine_turn_signal}")

                # 显示九转保护时间信息
                last_buy_time = self.nine_turn_signal_time[stock_code]["last_buy_time"]
                if last_buy_time is not None:
                    # 在回测模式下，使用数据时间计算保护时间
                    if self.backtest_mode and timestamp is not None:
                        try:
                            if isinstance(timestamp, (int, float)):
                                current_data_time = datetime.fromtimestamp(timestamp / 1000.0)
                            elif isinstance(timestamp, str):
                                current_data_time = pd.to_datetime(timestamp)
                            else:
                                current_data_time = timestamp.to_pydatetime() if hasattr(timestamp, 'to_pydatetime') else timestamp

                            time_diff = (current_data_time - last_buy_time).total_seconds() / 60
                        except Exception:
                            # 如果时间解析失败，使用当前时间
                            time_diff = (datetime.now() - last_buy_time).total_seconds() / 60
                    else:
                        # 实时模式使用当前时间
                        time_diff = (datetime.now() - last_buy_time).total_seconds() / 60

                    remaining_time = max(0, self.nine_turn_buy_protection_minutes - time_diff)
                    if remaining_time > 0:
                        self.logger.log_msg(f"   九转保护: 距离下次可买入还需 {remaining_time:.1f} 分钟")
                    else:
                        self.logger.log_msg(f"   九转保护: 可以买入")
            else:
                self.logger.log_msg(f"   九转: 无信号")

            ##显示macd信号
            macd = indicators.get('macd')
            self.logger.log_msg(f"   MACD: {macd}")
            

        except Exception as e:
            self.logger.log_msg(f"⚠️ {stock_code} 指标打印异常: {e}")
            import traceback
            traceback.print_exc()



    ##触发买卖
    def check_buy_sell(self, stock_code: str, stock_name: str, last_price: float, indicators: dict, timestamp=None):
        """根据指标结果判断是否发出买入卖出信号"""
        need_buy = False
        need_sell = False

        # 获取指标值
        rsi_value = indicators.get('rsi')
        macd = indicators.get('macd', '')
        atr_ratio = indicators.get('atr_ratio')
        vma = indicators.get('vma', False)

        # RSI超卖信号
        if rsi_value is not None and 10 >= rsi_value > 1:
            show_notification_thread("RSI买提示", f"{stock_name}RSI买")
            need_buy = True
            self.logger.log_signal(f"RSI超卖信号触发：{stock_code} RSI={rsi_value:.2f}")

        # MACD金叉信号
        if macd == "金叉":
            # MACD计算器已经处理了信号去重，这里直接处理业务逻辑
            self.logger.log_signal(f"{stock_name} 出现金叉,对比ATR，RSI，VMA")

            can_buy = 0

            # ATR波动率过滤
            if atr_ratio is not None and atr_ratio > 0:
                threshold = max(0.0006, atr_ratio * 0.7)  # 动态阈值
                if atr_ratio > threshold:
                    can_buy += 1
                else:
                    self.logger.log_signal(
                        f"ATR波动率不符合：{stock_code} {last_price} {atr_ratio:.6f} {threshold:.6f}")
            else:
                # 无法获取ATR时，默认允许
                can_buy += 1

            # RSI过滤
            if rsi_value is not None and rsi_value > 0:
                if rsi_value < 30:
                    can_buy += 1
                else:
                    self.logger.log_signal(f"RSI不符合：{stock_code}:{rsi_value:.2f}")

            # VMA过滤
            if vma:
                can_buy += 1
            else:
                self.logger.log_signal(f"VMA不符合：{stock_code}")

            # 三个条件都满足才买入
            if can_buy == 3:
                need_buy = True
                try:
                    atr_ratio_str = f"{atr_ratio:.6f}" if atr_ratio is not None else "NA"
                    self.logger.log_signal(
                        f"买入触发：{stock_code} price={last_price} macd=金叉 rsi={rsi_value:.2f} atr_ratio={atr_ratio_str} vma={vma}")
                except Exception:
                    pass

        # 九转买入信号
        nine_turn_signal = indicators.get('nine_turn', '')
        if nine_turn_signal == "买入":
            # 在回测模式下，使用tick数据的时间；实时模式下使用当前时间
            if self.backtest_mode and timestamp is not None:
                try:
                    if isinstance(timestamp, (int, float)):
                        current_time = datetime.fromtimestamp(timestamp / 1000.0)
                    elif isinstance(timestamp, str):
                        current_time = pd.to_datetime(timestamp).to_pydatetime()
                    else:
                        current_time = timestamp.to_pydatetime() if hasattr(timestamp, 'to_pydatetime') else timestamp
                except Exception:
                    current_time = datetime.now()
            else:
                current_time = datetime.now()

            # 1. 检查是否与上次九转买入信号相同，避免重复提示和下单
            current_signal_key = f"九转买_{last_price}"
            last_signal = self.signal_status[stock_code]["last_nine_turn_buy"]

            if current_signal_key == last_signal:
                # 这是重复的九转买入信号，不执行操作
                return

            # 2. 检查重复买入保护时间（10分钟内不重复买入）
            last_buy_time = self.nine_turn_signal_time[stock_code]["last_buy_time"]
            if last_buy_time is not None:
                time_diff = (current_time - last_buy_time).total_seconds() / 60  # 转换为分钟
                if time_diff < self.nine_turn_buy_protection_minutes:
                    self.log(f"⏰ {stock_code} 九转买入信号被保护时间过滤：距离上次买入仅 {time_diff:.1f} 分钟，需要等待 {self.nine_turn_buy_protection_minutes} 分钟")
                    return

            # 3. 检查信号是否是最新的（避免重启后获取到过期信号）
            # 获取九转计算器的缓存时间信息
            try:
                analysis = self.nine_turn.get_nine_turn_analysis(stock_code)
                if analysis.get('cached', False):
                    # 如果是缓存数据，检查缓存时间
                    cache_time = getattr(self.nine_turn, '_cache_time', {}).get(stock_code)
                    if cache_time is not None:
                        cache_age_minutes = (current_time.timestamp() - cache_time) / 60
                        if cache_age_minutes > 2:  # 缓存超过2分钟，认为可能过期
                            self.log(f"⚠️ {stock_code} 九转信号可能过期：缓存时间 {cache_age_minutes:.1f} 分钟")
                            # 强制刷新数据
                            self.nine_turn.reset(stock_code)
                            # 重新获取信号
                            if self.backtest_mode and current_time:
                                nine_turn_signal = self.nine_turn.get_previous_minute_signal(stock_code, end_time=current_time)
                            else:
                                nine_turn_signal = self.nine_turn.get_previous_minute_signal(stock_code)
                            if nine_turn_signal != "买入":
                                self.log(f"🔄 {stock_code} 刷新后九转信号变为: {nine_turn_signal}")
                                return
            except Exception as e:
                self.log(f"⚠️ {stock_code} 检查九转信号时间异常: {e}")

            # 4. 通过所有检查，执行买入操作
            self.signal_status[stock_code]["last_nine_turn_buy"] = current_signal_key
            self.nine_turn_signal_time[stock_code]["last_buy_time"] = current_time

            show_notification_thread("九转提示", f"{stock_name}九转买")
            need_buy = True
            try:
                # 在回测模式下，显示tick数据的时间；实时模式下显示当前时间
                if self.backtest_mode and timestamp is not None:
                    try:
                        if isinstance(timestamp, (int, float)):
                            display_time = datetime.fromtimestamp(timestamp / 1000.0).strftime('%H:%M:%S')
                        elif isinstance(timestamp, str):
                            display_time = pd.to_datetime(timestamp).strftime('%H:%M:%S')
                        else:
                            display_time = timestamp.strftime('%H:%M:%S') if hasattr(timestamp, 'strftime') else str(timestamp)
                    except Exception:
                        display_time = current_time.strftime('%H:%M:%S')
                else:
                    display_time = current_time.strftime('%H:%M:%S')

                self.logger.log_signal(f"买入触发：{stock_code} 九转买 price={last_price} 时间: {display_time}")
            except Exception:
                pass

        # 九转卖出信号
        nine_turn_signal = indicators.get('nine_turn', '')
        if nine_turn_signal == "卖出":
            # 在回测模式下，使用tick数据的时间；实时模式下使用当前时间
            if self.backtest_mode and timestamp is not None:
                try:
                    if isinstance(timestamp, (int, float)):
                        current_time = datetime.fromtimestamp(timestamp / 1000.0)
                    elif isinstance(timestamp, str):
                        current_time = pd.to_datetime(timestamp).to_pydatetime()
                    else:
                        current_time = timestamp.to_pydatetime() if hasattr(timestamp, 'to_pydatetime') else timestamp
                except Exception:
                    current_time = datetime.now()
            else:
                current_time = datetime.now()

            # 1. 检查是否与上次九转卖出信号相同，避免重复提示和下单
            current_signal_key = f"九转卖_{last_price}"
            last_signal = self.signal_status[stock_code]["last_nine_turn_sell"]

            if current_signal_key == last_signal:
                # 这是重复的九转卖出信号，不执行操作
                return

            # 2. 检查信号是否是最新的（避免重启后获取到过期信号）
            try:
                analysis = self.nine_turn.get_nine_turn_analysis(stock_code)
                if analysis.get('cached', False):
                    # 如果是缓存数据，检查缓存时间
                    cache_time = getattr(self.nine_turn, '_cache_time', {}).get(stock_code)
                    if cache_time is not None:
                        cache_age_minutes = (current_time.timestamp() - cache_time) / 60
                        if cache_age_minutes > 2:  # 缓存超过2分钟，认为可能过期
                            self.log(f"⚠️ {stock_code} 九转卖出信号可能过期：缓存时间 {cache_age_minutes:.1f} 分钟")
                            # 强制刷新数据
                            self.nine_turn.reset(stock_code)
                            # 重新获取信号
                            if self.backtest_mode and current_time:
                                nine_turn_signal = self.nine_turn.get_previous_minute_signal(stock_code, end_time=current_time)
                            else:
                                nine_turn_signal = self.nine_turn.get_previous_minute_signal(stock_code)
                            if nine_turn_signal != "卖出":
                                self.log(f"🔄 {stock_code} 刷新后九转信号变为: {nine_turn_signal}")
                                return
            except Exception as e:
                self.log(f"⚠️ {stock_code} 检查九转卖出信号时间异常: {e}")

            # 3. 通过所有检查，执行卖出操作
            self.signal_status[stock_code]["last_nine_turn_sell"] = current_signal_key
            self.nine_turn_signal_time[stock_code]["last_sell_time"] = current_time

            need_sell = True
            try:
                # 在回测模式下，显示tick数据的时间；实时模式下显示当前时间
                if self.backtest_mode and timestamp is not None:
                    try:
                        if isinstance(timestamp, (int, float)):
                            display_time = datetime.fromtimestamp(timestamp / 1000.0).strftime('%H:%M:%S')
                        elif isinstance(timestamp, str):
                            display_time = pd.to_datetime(timestamp).strftime('%H:%M:%S')
                        else:
                            display_time = timestamp.strftime('%H:%M:%S') if hasattr(timestamp, 'strftime') else str(timestamp)
                    except Exception:
                        display_time = current_time.strftime('%H:%M:%S')
                else:
                    display_time = current_time.strftime('%H:%M:%S')

                self.logger.log_signal(f"九转卖出信号触发：{stock_code} price={last_price} 时间: {display_time}")
            except Exception:
                pass

        # 死叉信号处理
        if macd == "死叉":
            # MACD计算器已经处理了信号去重，这里直接处理业务逻辑
            need_sell = True
            try:
                self.logger.log_signal(f"死叉信号触发：{stock_code} price={last_price}")
            except Exception:
                pass

        # 执行交易操作
        if need_sell:
            self.execute_death_cross_sell(stock_code)
        if need_buy:
            self.execute_golden_cross_buy(stock_code, stock_name, last_price)
    ##买入
    def execute_golden_cross_buy(self, stock_code: str, stock_name: str, current_price: float) -> bool:
        try:
            # 检查是否在交易时间段内
            if not self._is_trading_time():
                self.logger.log_signal("不在交易时间内")
                return False
            # 执行金叉自动买入
            try:
                order_data = {
                    "type": "buy",
                    "code": stock_code,
                    "order_remark": "MACD金叉自动买入"
                }
                self.saver.publish_order(order_data=order_data)
                return True
            except Exception as e:
                print(f"❌ 发送交易指令到队列失败: {e}")
                return False

        except Exception as e:
            self.log(f"❌ 执行金叉自动买入异常: {e}")
            return False
    ##卖出
    def execute_death_cross_sell(self, stock_code: str) -> bool:
        try:
            # 检查是否在交易时间段内
            if not self._is_trading_time():
                self.logger.log_signal("不在交易时间内")
                return False
                
            # 执行死叉自动卖出
            try:
                order_data = {
                    "type": "sell",
                    "code": stock_code,
                    "order_remark": "MACD死叉自动卖出"
                }
                self.saver.publish_order(order_data=order_data)
                return True
            except Exception as e:
                print(f"❌ 发送卖出指令到队列失败: {e}")
                return False
                
        except Exception as e:
            self.log(f"❌ 执行死叉自动卖出异常: {e}")
            return False
    ##交易时间
    def _is_trading_time(self) -> bool:
        """使用data_processor检查当前是否在交易时间段内"""
        return data_processor.is_trading_time()


    ##前端日志
    def log(self, msg):
        """统一的日志记录方法：同时输出到控制台和文件"""
        print(msg)
    ##获取最新的tick
    def _get_latest_tick_data(self, stock_data):
        """
        获取最新的tick数据（按时间戳排序取最大值）

        Args:
            stock_data: tick数据，可能是单个dict或list

        Returns:
            dict: 最新的tick数据
        """
        try:
            if isinstance(stock_data, list) and stock_data:
                def _as_ts(x):
                    try:
                        t = x.get('time') if isinstance(x, dict) else None
                        if t is None:
                            return -1.0
                        if isinstance(t, (int, float)):
                            return float(t)
                        s = str(t)
                        if s.isdigit():
                            return float(s)
                        ts = pd.to_datetime(s, errors='coerce')
                        if pd.isna(ts):
                            return -1.0
                        return float(ts.timestamp() * 1000.0)
                    except Exception:
                        return -1.0

                # 返回时间戳最大的tick数据
                return max(stock_data, key=_as_ts)
            else:
                # 如果不是列表或者为空，返回第一个元素
                return stock_data[0] if stock_data else {}
        except Exception:
            # 异常情况下返回第一个元素
            return stock_data[0] if stock_data else {}
    ##启动
    def start(self):
        if self.backtest_mode:
            self._start_backtest_mode()
        else:
            self._start_realtime_mode()
    def _start_realtime_mode(self):
        """启动实时模式"""
        self.logger.log_signal(f"开启实时行情")
        self._running = True

        # 订阅实时行情（tick级别）
        for code in self.stock_codes:
            self.xtdata.subscribe_quote(
                code,
                period=self.period,
                count=-1,
                callback=self.quote_callback
            )
            self.log(f"已订阅 {code} 的tick级别实时行情")

        # RSI计算器现在完全独立，不需要订阅分钟数据

        self.log(f"开始监听 {len(self.stock_codes)} 只股票的实时行情...")

        try:
            self.xtdata.run()  # 阻塞
        except KeyboardInterrupt:
            self.log("行情监控已手动停止")
        except Exception as e:
            self.log(f"行情监控异常: {e}")
    def _start_backtest_mode(self):
        """启动回测模式"""
        self.logger.log_signal(f"开启回测模式 - 日期: {self.backtest_date}")
        self._running = True

        print(f"🔄 正在启动回测模式...")
        print(f"  - 回测日期: {self.backtest_date}")
        print(f"  - 回测股票: {', '.join(self.stock_codes)}")

        # 准备历史数据
        self.backtest_manager.prepare_historical_data()

        # 检查是否有可用数据
        if not self.backtest_manager.has_more_data():
            self.log("❌ 没有可用的回测数据")
            return

        print("✅ 回测数据准备完成，开始模拟tick推送...")
        print("💡 按 Ctrl+C 可以停止回测")
        print("-" * 50)

        try:
            # 开始回测循环
            self._run_backtest_loop()
        except KeyboardInterrupt:
            self.log("回测已手动停止")
        except Exception as e:
            self.log(f"回测异常: {e}")
            import traceback
            traceback.print_exc()
    def _run_backtest_loop(self):
        """运行回测循环（优化版，提高回测速度）"""
        tick_count = 0
        start_time = time.time()

        # 根据快速模式调整批处理大小
        if self.backtest_manager.fast_mode:
            batch_size = 100  # 快速模式：每批处理100个tick
            progress_interval = 200  # 每200个tick显示一次进度
        else:
            batch_size = 10   # 普通模式：每批处理10个tick
            progress_interval = 50   # 每50个tick显示一次进度

        print(f"🚀 回测配置: 批处理大小={batch_size}, 进度显示间隔={progress_interval}")

        while self._running and self.backtest_manager.has_more_data():
            try:
                # 获取下一批tick数据（批量获取）
                tick_batch = self.backtest_manager.get_next_tick_batch(batch_size)

                if not tick_batch:
                    # 没有更多数据，结束回测
                    break

                # 批量处理tick数据
                for stock_code, ticks in tick_batch.items():
                    for tick in ticks:
                        # 模拟tick回调
                        self.quote_callback({stock_code: [tick]})
                        tick_count += 1

                # 显示进度（根据模式调整显示频率）
                if tick_count % progress_interval == 0:
                    progress = self.backtest_manager.get_progress()
                    progress_str = ", ".join([f"{code}: {p:.1f}%" for code, p in progress.items()])
                    elapsed_time = time.time() - start_time
                    ticks_per_second = tick_count / elapsed_time if elapsed_time > 0 else 0
                    remaining_ticks = sum(len(data) - pos for data, pos in zip(
                        self.backtest_manager.historical_data.values(),
                        self.backtest_manager.current_positions.values()
                    ))
                    eta_seconds = remaining_ticks / ticks_per_second if ticks_per_second > 0 else 0
                    eta_minutes = eta_seconds / 60

                    print(f"📊 回测进度: {progress_str} | 速度: {ticks_per_second:.1f} tick/秒 | 剩余: {eta_minutes:.1f} 分钟")

                # 快速模式下无延迟，普通模式下可以添加小延迟
                if not self.backtest_manager.fast_mode:
                    time.sleep(0.001)  # 1毫秒延迟

            except Exception as e:
                self.log(f"❌ 回测循环异常: {e}")
                continue

        # 回测完成
        end_time = time.time()
        total_time = end_time - start_time
        print("\n🎯 回测完成！")
        print(f"  - 总tick数量: {tick_count}")
        print(f"  - 回测日期: {self.backtest_date}")
        print(f"  - 总耗时: {total_time:.2f} 秒")
        print(f"  - 平均速度: {tick_count/total_time:.1f} tick/秒")
        print("✅ 回测模式已结束")
        ##停止
    def stop(self):
        # 这里需要你根据xtdata的API实现停止订阅和run
        self._running = False
        self.log("行情监控已停止")

    ##macd
    ##检查MACD金叉死叉信号
    def check_macd_signals(self, stock_code: str, current_price: float):
        current_time = time.time()
        stock_name = self.stock_type.get(stock_code)

        # 检查缓存是否有效
        if (CacheConfig.should_use_cache() and
                stock_code in self.macd_cache and
                MACDConfig.is_cache_valid(self.last_macd_check[stock_code], current_time)):
            # 使用缓存的结果
            cached_result = self.macd_cache[stock_code]
            self.process_macd_signals(stock_code, stock_name, current_price, cached_result)
            return

        # 检查缓存大小，如果超过限制则清理
        if len(self.macd_cache) > self.max_cache_size:
            self._cleanup_cache()

        try:
            # 性能监控开始
            start_time = time.time()

            # 异步计算MACD，避免阻塞主线程
            if MACDConfig.ENABLE_DETAILED_LOGGING:
                self.log(f"🔄 {stock_name} 开始计算MACD信号...")

            # 获取MACD信号（参数与配置保持一致）
            macd_signals = self.macd_calculator.detect_macd_signals(
                stock_code,
                fast=self.macd_fast,
                slow=self.macd_slow,
                signal=self.macd_signal,
                use_warmup=MACDConfig.USE_WARMUP
            )

            # 性能监控结束
            calculation_time = time.time() - start_time

            # 记录性能指标
            if PerformanceMonitor.should_monitor_performance():
                self._log_performance_metrics(stock_code, calculation_time)

            # 缓存结果
            self.macd_cache[stock_code] = macd_signals
            self.last_macd_check[stock_code] = current_time

            # 处理MACD信号
            macd = self.process_macd_signals(stock_code, stock_name, current_price, macd_signals)

            return macd

        except Exception as e:
            self.log(f"❌ {stock_name} MACD计算异常: {e}")
            import traceback
            traceback.print_exc()

    ##清理缓存
    def _cleanup_cache(self):
        current_time = time.time()
        expired_keys = []

        for code, last_check in self.last_macd_check.items():
            if current_time - last_check > self.macd_cache_ttl:
                expired_keys.append(code)

        # 删除过期缓存
        for code in expired_keys:
            if code in self.macd_cache:
                del self.macd_cache[code]
            if code in self.last_macd_check:
                del self.last_macd_check[code]

        # 如果缓存仍然过大，删除最旧的条目
        if len(self.macd_cache) > self.max_cache_size:
            sorted_codes = sorted(self.last_macd_check.items(), key=lambda x: x[1])
            codes_to_remove = len(self.macd_cache) - self.max_cache_size

            for i in range(codes_to_remove):
                if i < len(sorted_codes):
                    code = sorted_codes[i][0]
                    if code in self.macd_cache:
                        del self.macd_cache[code]
                    if code in self.last_macd_check:
                        del self.last_macd_check[code]

    ##重置信号状态，清除历史记录
    def reset_signal_status(self, stock_code: str = None):
        if stock_code:
            # 重置指定股票的信号状态
            if stock_code in self.signal_status:
                self.signal_status[stock_code] = {"last_golden_cross": None, "last_death_cross": None}
        else:
            # 重置所有股票的信号状态
            for code in self.stock_codes:
                self.signal_status[code] = {"last_golden_cross": None, "last_death_cross": None}

    ##获取信号状态信息
    def get_signal_status(self, stock_code: str = None):
        if stock_code:
            return self.signal_status.get(stock_code, {})
        else:
            return self.signal_status

    ##记录性能指标
    def _log_performance_metrics(self, stock_code: str, calculation_time: float):
        thresholds = PerformanceMonitor.get_performance_thresholds()

        if calculation_time > thresholds['max_calculation_time']:
            self.log(f"⚠️ {stock_code} MACD计算时间过长: {calculation_time:.2f}秒")

        if PerformanceMonitor.LOG_PERFORMANCE_METRICS:
            self.log(f"📊 {stock_code} MACD计算耗时: {calculation_time:.3f}秒")

    ##处理MACD信号结果
    def process_macd_signals(self, stock_code: str, stock_name: str, current_price: float, macd_signals: Dict):
        try:
            # 检查金叉信号
            if macd_signals.get('golden_cross'):
                latest_golden = macd_signals['golden_cross'][-1]  # 最新的金叉信号
                signal_time = latest_golden.get('time', '')
                signal_price = latest_golden.get('price', 0)

                # 检查是否是新产生的金叉信号（时间较近）
                if signal_time:
                    try:
                        signal_datetime = datetime.strptime(signal_time, '%Y-%m-%d %H:%M:%S')
                        time_diff = abs((datetime.now() - signal_datetime).total_seconds())

                        # 如果金叉信号在最近指定时间内产生，则发出通知
                        signal_window = MACDConfig.get_signal_window()
                        if time_diff < signal_window:
                            # 检查是否与上次金叉信号相同，避免重复提示和下单
                            current_signal_key = f"{signal_time}_{signal_price}"
                            last_signal = self.signal_status[stock_code]["last_golden_cross"]

                            if current_signal_key != last_signal:
                                # 这是新的金叉信号，执行操作
                                self.signal_status[stock_code]["last_golden_cross"] = current_signal_key

                                # 调用封装的买入方法
                                return "金叉"
                            else:
                                # 这是重复的金叉信号，不执行操作
                                return None
                    except Exception as e:
                        self.log(f"❌ 处理金叉信号时间异常: {e}")
                        pass

            # 检查死叉信号
            if macd_signals.get('death_cross'):
                latest_death = macd_signals['death_cross'][-1]  # 最新的死叉信号
                signal_time = latest_death.get('time', '')
                signal_price = latest_death.get('price', 0)

                # 检查是否是新产生的死叉信号（时间较近）
                if signal_time:
                    try:
                        signal_datetime = datetime.strptime(signal_time, '%Y-%m-%d %H:%M:%S')
                        time_diff = abs((datetime.now() - signal_datetime).total_seconds())

                        # 如果死叉信号在最近指定时间内产生，则发出通知
                        signal_window = MACDConfig.get_signal_window()
                        if time_diff < signal_window:
                            # 检查是否与上次死叉信号相同，避免重复提示
                            current_signal_key = f"{signal_time}_{signal_price}"
                            last_signal = self.signal_status[stock_code]["last_death_cross"]

                            if current_signal_key != last_signal:
                                # 这是新的死叉信号，执行操作
                                self.signal_status[stock_code]["last_death_cross"] = current_signal_key

                                msg = f"{stock_name}死叉"
                                self.log(msg)
                                show_notification_thread("MACD死叉", msg)
                                # 调用封装的按代码平仓
                                return "死叉"
                            else:
                                # 这是重复的死叉信号，不执行操作
                                return None
                    except Exception:
                        pass

            # 输出最新MACD状态（减少输出频率，只在有变化时输出）
            latest_macd = macd_signals.get('latest_macd', {})
            if latest_macd:
                diff = latest_macd.get('diff', 0)
                dea = latest_macd.get('dea', 0)
                trend = latest_macd.get('trend', 'unknown')

                # 只在趋势变化时输出
                if stock_code not in self.macd_cache.get('last_trend', {}):
                    self.macd_cache['last_trend'] = {}

                if self.macd_cache['last_trend'].get(stock_code) != trend:
                    self.macd_cache['last_trend'][stock_code] = trend
                    self.log(f"📊 {stock_name} MACD趋势: {trend}, DIFF: {diff:.6f}, DEA: {dea:.6f}")

        except Exception as e:
            self.log(f"❌ {stock_name} 处理MACD信号异常: {e}")


# ==================== 直接启动代码 ====================
if __name__ == "__main__":

    backtest = True
    stocks = "513120.SH"
    date = "20250822"
    try:
        if backtest:

            # 解析股票代码
            backtest_stocks = [code.strip() for code in stocks.split(',')]

            print("🔄 正在启动回测模式...")
            print(f"  - 回测日期: {date}")
            print(f"  - 回测股票: {', '.join(backtest_stocks)}")

            # 创建回测监控实例
            monitor = MarketMonitor(
                backtest_mode=True,
                backtest_stocks=backtest_stocks,
                backtest_date=date
            )

        else:
            # 实时模式
            print("🚀 正在启动实时行情监控...")

            # 创建实时监控实例
            monitor = MarketMonitor()

        print("✅ 监控实例创建成功")
        print("📊 监控股票列表:")
        for code in monitor.stock_codes:
            name = monitor.stock_type.get(code, "未知")
            print(f"   {code} - {name}")

        print("\n🔥 开始启动监控...")
        print("💡 按 Ctrl+C 可以停止监控")
        print("-" * 50)

        # 启动监控（这会阻塞直到手动停止）
        monitor.start()

    except KeyboardInterrupt:
        print("\n⏹️  收到停止信号，正在关闭监控...")
        try:
            monitor.stop()
        except:
            pass
        print("✅ 监控已停止")
    except Exception as e:
        print(f"❌ 启动监控失败: {e}")
        import traceback
        traceback.print_exc()
        exit(1)